package com.superhero.lock.util.tool;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Function;

/**
 *并行帮助类
 *
 *@author weijianxun
 *@date 2023/11/30 10:04
 */
public class ParallelUtil {
    /**
     * 获取异步执行结果，默认线程池且调用完释放
     * @param dataList
     * @param function
     * @return
     * @param <T>
     * @param <R>
     */
    public static <T, R> List<R> autoGetSubmit(List<T> dataList, Function<T, R> function) {
        ThreadPoolTaskExecutor taskExecutor = initPool(3, 5, "autoGetSubmit-");
        return autoGetSubmitAndClosePool(dataList, function, taskExecutor);
    }

    /**
     * 获取异步执行结果,线程池释放
     * @param dataList
     * @param function
     * @param taskExecutor
     * @return
     * @param <T>
     * @param <R>
     */
    public static <T, R> List<R> autoGetSubmitAndClosePool(List<T> dataList, Function<T, R> function, ThreadPoolTaskExecutor taskExecutor) {
        List<R> result = autoGetSubmitAndNoClosePool(dataList, function, taskExecutor);
        destroyPool(taskExecutor);
        return result;
    }

    /**
     * 获取异步执行结果，线程池不释放
     * @param dataList
     * @param function
     * @param taskExecutor
     * @return
     * @param <T>
     * @param <R>
     */
    public static <T, R> List<R> autoGetSubmitAndNoClosePool(List<T> dataList, Function<T, R> function, ThreadPoolTaskExecutor taskExecutor) {
        List<Future<R>> submit = submit(dataList, function, taskExecutor);
        List<R> result = waitGetSubmit(submit);
        return result;
    }


    /**
     * 初始化线程池
     * @param corePoolSize
     * @param maxPoolSize
     * @param threadNamePrefixSet
     * @return
     */
    public static ThreadPoolTaskExecutor initPool(int corePoolSize, int maxPoolSize, String threadNamePrefixSet) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix(threadNamePrefixSet + "-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

    /**
     * 销毁线程池
     * @param taskExecutor
     */
    public static void destroyPool(ThreadPoolTaskExecutor taskExecutor) {
        taskExecutor.shutdown();
    }

    /**
     * 提交任务
     * @param dataList 需要并行处理的任务
     * @param function 核心操作逻辑
     * @param taskExecutor 线程池
     * @return
     * @param <T>
     * @param <R>
     */
    public static <T, R> List<Future<R>> submit(List<T> dataList, Function<T, R> function, ThreadPoolTaskExecutor taskExecutor) {
        List<Future<R>> result = new ArrayList<>();
        for (T data : dataList) {
            Future<R> submit = taskExecutor.submit(() -> function.apply(data));
            result.add(submit);
        }
        return result;
    }

    /**
     * 等待所有任务完成并获取结果
     * @param futures
     * @param <R>
     * @throws Exception
     */
    public static <R> List<R> waitGetSubmit(List<Future<R>> futures) {
        List<R> result = new ArrayList<>();
        for (Future<R> future : futures) {
            try {
                R data = future.get();
                result.add(data);
            } catch (Exception e) {
                // todo log error
                continue;
            }
        }
        return result;
    }

}
